package com.se.stream;

import com.model.Order;
import lombok.extern.log4j.Log4j2;
import reactor.core.publisher.Flux;

import java.util.*;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.*;

import static java.util.stream.Collectors.*;

/**
 * @Author: hdu_huang
 * @Date: 2019/1/15 14:28
 */
@Log4j2
public class StreamTest {

    //Enumeration Iterable,iterator,Spliterator
    public static void main(String[] args) {

//        createStream();
//        baseStream();
//        spliterator();
//        iterator();
//        optStream();
//        collectStream();
        collectHighStream();
    }

    /**
     * @author hdu_huang
     * @date 2019/1/18
     * @funs Stream 核心是 Spliterator  Spliterators[ite <-> split], StreamSupport[split->Stream]
     * Stream只能消费[终止操作]一次
     */
    private static void createStream() {

        //Stream static method [of, empty, ofNullable[9], iterate*, generate] 连接 concat
        //IntStream/LongStream [range, rangeClosed]  Stream.builder()
        //Collection[最常用] spliterator/stream,parallelStream
        //other Files/BufferedReader.lines  BitSet.stream String[chars,codePoints]
        //util Arrays.stream  StreamSupport

        List<Integer> integers = Arrays.asList(4, 2, 4, 1, 6, 3, 5);
        int[] arr = {4, 3, 6, 2};

        Stream.of("hello", "world");
        Stream.of(); // => empty()
        Stream.empty();
        Stream.ofNullable(null);

        Stream.generate(Math::random).limit(10); //infinite sequential @see notes

        Stream.iterate(0, integer -> integer + 1).limit(10); // [0,9]
        Stream.iterate(0, integer -> integer < 10, integer -> integer + 1); //[0,9] since 9

        Stream.concat(Stream.of(5, 2), integers.stream());

        Stream.builder().add("hello").add("world").build();

        IntStream.range(1, 100); // [1,100) see LongStream
        IntStream.rangeClosed(1, 100); // [1,100]

        integers.stream();
        Arrays.stream(arr);
        "hello world!".chars();
        StreamSupport.stream(integers.spliterator(), false);

        Flux.just("hello", "world").toStream(); //flux<->stream

        //flatMap 可以把元素增多[唯一] 返回Stream类型
        Stream<Stream<Integer>> streamStream = Stream.of(Stream.of(4, 3, 1), integers.stream());
        List<Integer> collect = Stream.of(streamStream).flatMap(Function.identity()).flatMap(Function.identity()).collect(toList());
        System.out.println(collect);

    }

    private static void baseStream() {
        Supplier<Stream<String>> supplier = () -> Stream.of("hello", "world!", "google", "hello world!");

        Iterator<String> iterator = supplier.get().iterator();

        Spliterator<String> spliterator = supplier.get().spliterator(); //只能遍历一次

        Stream<String> parallel = supplier.get().parallel(); //ForkJoinPool.commonPool-worker-n

        Stream<String> stringStream = supplier.get().onClose(() -> log.error("Stream onClose!")).onClose(() -> log.error("Stream onClose!!!"));

        try (stringStream) { // or stringStream.close();

        }

        Stream<Integer> sequential = parallel.map(String::length).sequential(); // serial Stream

        boolean isParallel = supplier.get().isParallel(); //false
        boolean isParallel1 = supplier.get().parallel().isParallel(); //true  fork-join

        log.info(sequential);
    }


    // takeWhile, dropWhile since 9
    // takeWhile 第一个不匹配元素前面后面所有元素[不包括不匹配的那个元素]
    // dropWhile 第一个不匹配元素后面所有元素  ****
    private static void optStream() { //sort 操作 一次遍历无法完成

        Supplier<Stream<String>> supplier = () -> Stream.of("hello", "world", "google", "hello world!", "bai-du");
        //中间操作 Stateless   Stateful[distinct/sorted/limit/skip]

        //findAny findFirst 并行上有差异
        IntSummaryStatistics intStat = supplier.get().map(String::length).map(e -> e * 2).skip(1).limit(3).mapToInt(Integer::intValue).summaryStatistics();

        List<String> strings = supplier.get().filter(s -> s.length() > 4).skip(1).limit(2).sorted(Comparator.comparing(String::length).reversed()).collect(toList());

        List<Integer> list = Arrays.asList(1, 3, 6, 4, 2, 8, 5);

        // dropWhile  第一个不匹配的元素 后面所有元素
        list.stream().dropWhile(integer -> integer > 4); //第一个不匹配元素 1 [1, 3, 6, 4, 2, 8, 5]
        list.stream().dropWhile(integer -> integer < 4); //第一个不匹配元素6 [6, 4, 2, 8, 5]

        // takeWhile < 4[1,3]   > 4[]
        List<Integer> takes = list.stream().takeWhile(integer -> integer < 4).collect(toList());

        list.stream().takeWhile(integer -> integer >= 4); //第一个不匹配1，[]

        log.info(takes);


        //forEach forEachOrdered[平行时候，只用一个额外线程]
        IntStream.rangeClosed(1, 20000).parallel();

        boolean bool = supplier.get().flatMap(s -> Stream.of(s.split(""))).distinct().anyMatch(s -> s.equals("o"));

        //aggregation reduce(min,max,count[不触发peek]) *reduce*
        long min = list.stream().min(Integer::compareTo).get();
        long max = list.stream().max(Integer::compareTo).get();
        long count = list.stream().peek(integer -> log.info(integer)).count();
        //reduce
        long sum = list.stream().reduce(Integer::sum).orElse(0);
        int sum1 = list.stream().reduce(0, Integer::sum);

        //此reduce可取代部分collect [不一定是容器]  ints == reduce  true
        int sums = list.stream().map(e -> 3 * e).reduce(0, (int1, int2) -> int1 + int2, (int1, int2) -> int1 + int2);

        ArrayList<Integer> ints = new ArrayList<>(list.size());
        ArrayList<Integer> toList = list.stream().map(e -> 2 * e).reduce(ints, (integers, integer) -> {
            integers.add(integer);
            return integers;
        }, (integers, integers2) -> {
            integers.addAll(integers2);
            return integers;
        });

        Map<String, Integer> map = new HashMap<>();
        supplier.get().flatMap(s -> Stream.of(s.split(""))).reduce(map, (map1, s) -> {
            map1.put(s, map1.containsKey(s) ? (map1.get(s) + 1) : 1);
            return map1;
        }, (map1, map2) -> map1);


        log.info(toList);
        log.info(map);

        boolean anyMatch = list.stream().anyMatch(integer -> integer > 7); //true
        boolean allMatch = list.stream().allMatch(integer -> integer > 1); //false
        boolean noneMatch = list.stream().noneMatch(integer -> integer > 8); //true

        // 并行下不一样, @see forEach
        Integer findFirst = list.stream().parallel().findFirst().get();
        Integer findAny = list.stream().parallel().findAny().get();

        log.info("");

    }

    // takeWhile, dropWhile
    private static void collectStream() {

        Supplier<Stream<String>> supplier = () -> Stream.of("hello", "world!", "google", "hello world!", "baidu");

        Map<String, Long> collect = supplier.get().flatMap(s -> Stream.of(s.split("")))
                .collect(Collectors.groupingBy(s -> s, Collectors.mapping(o -> {
                    return o;
                }, Collectors.counting())));

        //没有 Function<A, R> finisher() 方法
        long[] sums = collect.values().stream().collect(() -> new long[1], (longs, aLong) -> longs[0] += aLong, (longs, longs2) -> longs[0] += longs2[0]);
        long[] sums1 = collect.values().stream().collect(Collector.of(() -> new long[1], (longs, aLong) -> longs[0] += aLong, (longs, longs2) -> longs));

        // Collector.of  有 finisher 方法  容器 -> 值
        long sum = collect.values().stream().collect(Collector.of(() -> new long[1], (longs, aLong) -> longs[0] += aLong, (longs, longs2) -> longs, longs -> longs[0]));
        long sum1 = collect.values().stream().collect(Collector.of(() -> new long[1], (longs, aLong) -> longs[0] += aLong, (longs, longs2) -> longs, longs -> longs[0], Collector.Characteristics.CONCURRENT));

        // Collectors 模板方法 [important]
        List<Integer> ints = List.of(1, 4, 3, 6, 3, 5);

        //转为集合 Collection(list,set), Map
        Function<Integer, Integer> dou = e -> e * 2;
        HashSet<Integer> set = ints.stream().map(dou).collect(toCollection(HashSet::new));
        set.add(13);

        //不可修改 Collections.unmodifiable*  since10
        Set<Integer> set1 = ints.stream().map(dou).collect(toSet());
        Set<Integer> set2 = ints.stream().map(dou).collect(toUnmodifiableSet());

        ints.stream().map(dou).collect(toList());
        ints.stream().map(dou).collect(toUnmodifiableSet());

        //important 转为map,unmodifiableMap,concurrentMap 1.key必须唯一  uniqKeysMapAccumulator     2.value合并
        Map<Integer, Integer> map = ints.stream().map(dou).distinct().collect(toMap(Function.identity(), dou));
        Map<Integer, Integer> map1 = ints.stream().map(dou).collect(toMap(Function.identity(), dou, (int1, int2) -> int1 + int2));
        ints.stream().map(dou).collect(toMap(Function.identity(), dou, (int1, int2) -> int1 + int2, TreeMap::new));

        ints.stream().map(dou).distinct().collect(toUnmodifiableMap(Function.identity(), dou));
        Map<Integer, Integer> map2 = ints.stream().map(dou).collect(toUnmodifiableMap(Function.identity(), dou, (int1, int2) -> int1 + int2));

        ints.stream().map(dou).distinct().collect(toConcurrentMap(Function.identity(), dou));

        ints.stream().map(dou).collect(toList());
        ints.stream().map(dou).collect(toUnmodifiableList());

        //统计相关
        ints.stream().map(dou).collect(counting());
        ints.stream().map(dou).collect(minBy(Integer::compareTo));
        ints.stream().map(dou).collect(maxBy(Comparator.comparingInt(Integer::intValue)));
        ints.stream().collect(averagingInt(Integer::intValue));
        IntSummaryStatistics sumStats = ints.stream().map(dou).collect(summarizingInt(Integer::intValue));
        ints.stream().map(dou).collect(reducing(0, Integer::sum));

        // 字符串拼接
        String str1 = supplier.get().collect(joining());
        String str2 = supplier.get().collect(Collectors.joining("-"));
        String str3 = supplier.get().collect(Collectors.joining("-", "A:", ":Z"));


        log.info(collect);
        log.info(map);
        log.info(str3);

    }

    //分组，分区， 下游收集器 downstream
    private static void collectHighStream() {

        Supplier<Stream<String>> supplier = () -> Stream.of("hello", "world!", "google", "helloWorld!", "baidu");

        List<Integer> list = List.of(1, 4, 3, 6, 3, 5, 3);
        Function<Integer, Integer> dou = e -> e * 2;

        list.stream().collect(mapping(dou, toList())); //=> map(dou).collect(toList())
        String str = supplier.get().collect(flatMapping(s -> Stream.of(s.split("")), mapping(s -> s, filtering(character -> true, joining()))));

        Map<String, Long> group = Stream.of(str.split("")).collect(Collectors.groupingBy(Function.identity(), counting()));

        Stream.of(str.split("")).collect(Collectors.groupingByConcurrent(Function.identity(), counting())); //ConcurrentMap
        Map<Boolean, List<String>> partition = Stream.of(str.split("")).collect(partitioningBy(s -> Character.valueOf(s.charAt(0)).compareTo('d') > 0));//

        Map<String, Long> group1 = supplier.get().flatMap(s -> Stream.of(s.split("")))
                .collect(Collectors.groupingBy(Function.identity(), counting()));

        Long count = list.stream().collect(collectingAndThen(counting(), aLong -> aLong * 2));

        //group, sort 高级应用
        Order order1 = new Order(1, "2019-02-01", 789, 2, 0, 10);
        Order order2 = new Order(2, "2019-02-01", 678, 1, 1, 10);
        Order order3 = new Order(3, "2019-02-02", 789, 1, 0, 9);
        Order order4 = new Order(4, "2019-02-02", 789, 3, 0, 10);
        Order order5 = new Order(5, "2019-02-03", 789, 1, 0, 8);

        Supplier<Stream<Order>> orderSupplier = () -> Stream.of(order1, order2, order3, order4, order5);

        Function<Order, String> groupBy = order -> order.getDate() + order.getUserId(); //...

        Map<String, List<Order>> collect = orderSupplier.get().collect(groupingBy(groupBy));


        log.info(collect);
        log.info(group);

    }


    private static void spliterator() {

        //Spliterator OfPrimitive  OfInt/OfLong/OfDouble -> tryAdvance,forEachRemaining,trySplit[二分]
        //Spliterators
        //Iterator ListIterator

        Spliterator<Integer> split0 = IntStream.rangeClosed(1, 300).boxed().collect(Collectors.toList()).spliterator(); //150-300

        //trySplit 二分 [可以使用线程池并行迭代]
        Spliterator<Integer> split1 = split0.trySplit(); // 75-150

        Spliterator<Integer> split2 = split1.trySplit(); // 37-75

        Spliterator<Integer> split3 = split2.trySplit(); // 0-37

        System.out.println(split3.estimateSize()); //37

        long size = split0.estimateSize();

        split0.tryAdvance(log::info);//forEachRemaining

        //自然排序 null,
//        Comparator<? super Integer> comparator = split0.getComparator(); //IllegalStateException

        long exactSizeIfKnown = split0.getExactSizeIfKnown();

        int characteristics = split0.characteristics();

        Spliterator<Integer> spliterator = Stream.iterate(1, integer -> 2 * integer).spliterator();

        long size1 = spliterator.estimateSize(); //Long.MAX_VALUE

        System.out.println(split3.estimateSize()); //37

    }

    private static void iterator() {

        //Iterator ListIterator[可以多次遍历]  forEachRemaining
        //PrimitiveIterator  OfInt/OfLong/OfDouble

        List<Integer> list = IntStream.rangeClosed(1, 10).boxed().collect(Collectors.toList());
        Iterator<Integer> it = list.iterator();

        Integer next = it.next();
//        it.forEachRemaining(log::info);

        ListIterator<Integer> listIt = list.listIterator();
        Integer next1 = listIt.next();
        Integer next2 = listIt.next();

        int i = listIt.previous();

        listIt.forEachRemaining(log::info);

        while (listIt.hasPrevious()) {
            log.info(listIt.previous());
        }
//        int i1 = listIt.previous();
        System.out.println(next);


    }

}
